package net.ziyoushu.demo1.context;

import net.ziyoushu.demo1.config.RocketMqProperties;
import net.ziyoushu.demo1.config.ScanRockerMqConsumerConfig;
import net.ziyoushu.demo1.annotation.RocketMqTag;
import net.ziyoushu.demo1.annotation.RocketMqTopic;
import net.ziyoushu.demo1.holder.AsynMessageConsumerHoler;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.lang.reflect.Method;
import java.util.*;

public class RocketMqConsumerContext {
    private static final Logger LOGGER = LoggerFactory.getLogger(RocketMqConsumerContext.class);

    /** topic 队列 **/
    private Map<String, Object> consumerMap = new HashMap<>();
    /** tag队列 **/
    private Map<Object, Map<String, Object>> consumerMethodMap = new HashMap<>();
    @Autowired
    RocketMqProperties rocketMqProperties;

    @PostConstruct
    public void scan() {
        ApplicationContext context = new AnnotationConfigApplicationContext(ScanRockerMqConsumerConfig.class);
        Map<String, Object> beansWithAnnotation = context.getBeansWithAnnotation(RocketMqTopic.class);
        Set<String> strings = beansWithAnnotation.keySet();
        for (String benName : strings) {
            Object bean = beansWithAnnotation.get(benName);
            Class<?> aClass = bean.getClass();
            RocketMqTopic annotation = aClass.getAnnotation(RocketMqTopic.class);
            String topic = annotation.topic();
            Method[] methods = aClass.getMethods();
            if(methods.length < 1) {
                continue;
            }

            Map<String, Object> consumerMethodItemMap = new HashMap<>();
            for (Method method : methods) {
                RocketMqTag methodConsumer = method.getAnnotation(RocketMqTag.class);
                if(null == methodConsumer) {
                    continue;
                }
                String tag = methodConsumer.tag();
                consumerMethodItemMap.put(tag, method);
            }
            if(consumerMethodItemMap.keySet().size() < 1){
                continue;
            }
            consumerMap.put(topic, beansWithAnnotation.get(benName));
            consumerMethodMap.put(bean, consumerMethodItemMap);
        }
    }

    /**
     * @Title 构造消费端
     * @Author lcb
     * @Date 2019/3/30
     **/
    @Bean
    public AsynMessageConsumerHoler getRocketMqConsumer() {
        AsynMessageConsumerHoler consumer = new AsynMessageConsumerHoler(rocketMqProperties.getConsumerGroupName());
        try {
            consumer.setNamesrvAddr(rocketMqProperties.getNamesrvAddr());
            // 消费顺序，指定从队列头到尾消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            Set<String> topics = consumerMap.keySet();
            if(CollectionUtils.isEmpty(topics)) {
                LOGGER.error("please set a topic at last, the system startup failed");
                System.exit(0);
            }
            for (String topic : topics) {
                consumer.subscribe(topic.trim(), "*");
            }

            //注册消息监听器
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                    MessageExt msg = msgs.get(0);
                    LOGGER.info("receive a message, msgId=" + msg.getMsgId());
                    try {
                        Object bean = consumerMap.get(msg.getTopic());
                        Map<String, Object> stringObjectMap = consumerMethodMap.get(bean);
                        Set<String> strings = stringObjectMap.keySet();

                        for (String tag : strings) {
                            if(tag.equals(msg.getTags())) {
                                Method method = (Method)stringObjectMap.get(tag);
                                method.setAccessible(true);
                                try {
                                    Object result = method.invoke(bean, msg);
                                    if(result instanceof Boolean) {
                                        return (Boolean) result ? ConsumeConcurrentlyStatus.CONSUME_SUCCESS : ConsumeConcurrentlyStatus.RECONSUME_LATER;
                                    }else {
                                        // 这里是指返回值不符合格式，实际上是已经消费了的
                                        LOGGER.error("禁止返回非布尔类型的值,如果出现此提示应该杀个程序员祭天");
                                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                                    }
                                }catch (Exception e) {
                                    LOGGER.error("***消费点已报错，自己去处理逻辑，消息会重发***");
                                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                                }

                            }
                        }
                    }catch (Exception e) {
                        LOGGER.info("failed to consum this message, msgId=" + msg.getMsgId());
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            });
            consumer.start();
            LOGGER.info("the consumer system startup success!");
        }catch (Exception e) {
            LOGGER.info("the system startup failed!");
        }
        return consumer;
    }
}
